package juc;

import java.util.Date;
import java.util.Scanner;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import java.util.stream.IntStream;

/**
 * 测试CyclicBarrier（循环屏障）
 * 基于Lock 和 Condition实现
 * 初始值 parties：参与方数量，不可更改，各个参与方 一起 参加多个任务。数量不可更改，但参与的线程是可以更改的。
 * barrier point: 所有使用方在此等待其它线程结束，然后开始新的操作。
 * @author ben
 * @date 2021-09-15 09:15:51 CST
 */
public class TestCyclicBarrier {

	public static Consumer<Object> cs = System.out::println;
	
	public static void main(String[] args) throws InterruptedException {

		cs.accept("请输入要执行的测试序号：");
		cs.accept("1、test1() 基本操作");
		cs.accept("2、test2() 异常操作");
		cs.accept("3、test3() await超时版");
		
		Scanner scann = new Scanner(System.in);
		while (true) {
			cs.accept("\n输入Q退出。\n请输入：");
			if (scann.hasNext()) {
				String ip = scann.next();
				if ("Q".equalsIgnoreCase(ip)) {
					cs.accept("Bye");
					scann.close();
					break;
				}
				
				try {
					int intp = Integer.parseInt(ip);
					if (intp == 1) {
						test1();
					} else if (intp == 2) {
						try {
							test2();
						} catch (InterruptedException e) {
							cs.accept("call test2(): InterruptedException");
						} catch (BrokenBarrierException e) {
							cs.accept("call test2(): BrokenBarrierException");
						}

						// 休眠1秒 等待其它线程 完全结束
						TimeUnit.SECONDS.sleep(1);
					} else if (intp == 3) {
						try {
							test3();
						} catch (InterruptedException e) {
							cs.accept("call test3(): InterruptedException");
						} catch (BrokenBarrierException e) {
							cs.accept("call test3(): BrokenBarrierException");
						}

						// 休眠1秒 等待其它线程 完全结束
						TimeUnit.SECONDS.sleep(3);
					} else {
						cs.accept("不支持的操作：" + ip);
					}
				} catch (NumberFormatException e) {
					cs.accept("错误输入，非选项：" + ip);
				}
			}
		}
		
		scann.close();
	}
	
	/**
	 * 试验await超时版（2参数）
	 * @author ben
	 * @date 2021-09-15 11:38:14 CST
	 * @throws InterruptedException
	 * @throws BrokenBarrierException
	 */
	public static void test3() throws InterruptedException, BrokenBarrierException {
		String tnameMain = Thread.currentThread().getName();
		cs.accept(tnameMain + ": start");

		// 工作线程数量
		final int workNum = 1;
		CyclicBarrier cb = new CyclicBarrier(workNum + 1);
		
		new Thread(()->{
			String tname = Thread.currentThread().getName();
			cs.accept(tname + ": start");
			
			// 小于主线程的 3：正常
//			long slt = 1;
			// 大于主线程的 3：异常
			long slt = 5;
			try {
				TimeUnit.SECONDS.sleep(slt);
			} catch (InterruptedException e1) {
				cs.accept(tname + ": InterruptedException 1");
			}
			
			try {
				cb.await();
			} catch (InterruptedException e) {
				cs.accept(tname + ": InterruptedException 2");
			} catch (BrokenBarrierException e) {
				// 抛出异常
				cs.accept(tname + ": BrokenBarrierException");
			}
			
			cs.accept(tname + ": end");
		}).start();
		
		try {
			cb.await(3, TimeUnit.SECONDS);
		} catch (InterruptedException e) {
			cs.accept(tnameMain + ": InterruptedException");
		} catch (BrokenBarrierException e) {
			cs.accept(tnameMain + ": BrokenBarrierException");
		} catch (TimeoutException e) {
			// 没有等到，发生超时异常
			cs.accept(tnameMain + ": TimeoutException");
		}
		
		cs.accept(tnameMain + ": end");
	}
	
	/**
	 * 异常操作
	 * @author ben
	 * @throws BrokenBarrierException 
	 * @throws InterruptedException 
	 * @date 2021-09-15 10:24:01 CST
	 */
	public static void test2() throws InterruptedException, BrokenBarrierException {
		String tnameMain = Thread.currentThread().getName();
		cs.accept(tnameMain + ": start");
		
		// 工作线程数量
		final int workNum = 2;

		// 初始化参与方数量，后续无法更改
		CyclicBarrier cb = new CyclicBarrier(workNum + 1);
		
		Thread[] tdarr = new Thread[workNum];
		cs.accept("tdarr.length=" + tdarr.length);
		
		//  启动工作线程
		IntStream.range(0, workNum).forEach(item->{
			Thread td = new Thread(()->{
				String tname = Thread.currentThread().getName();
				cs.accept(tname + ": start...now=" + new Date());
				
//				try {
//					int slt = ThreadLocalRandom.current().nextInt(10);
//					slt = 0;
//					cs.accept(tname + ": slt=" + slt);
//					TimeUnit.SECONDS.sleep(slt);
//				} catch (InterruptedException e) {
//					cs.accept(tname + ": InterruptedException 1...cb.isBroken=" + cb.isBroken() + ", parties=" + cb.getParties());
//					// 多次检查中断状态
//					cs.accept(tname + ": isInterrupted 1=" + Thread.currentThread().isInterrupted());
//					// 清理线程中断状态
//					Thread.interrupted();
////					Thread.interrupted();
//					cs.accept(tname + ": isInterrupted 2=" + Thread.currentThread().isInterrupted());
//				}
				
				// 被中断，中断之后，再等待？
				try {
					// 此时被中断，抛出InterruptedException，已等待线程执行 await()发生 BrokenBarrierException
					cs.accept(tname + ": 等待...now=" + new Date() + ", isBroken=" + cb.isBroken());
					cb.await();
				} catch (InterruptedException e) {
					cs.accept(tname + ": InterruptedException 2");
				} catch (BrokenBarrierException e) {
					// 等待时，cb执行reset()，发生此异常，任务结束
					cs.accept(tname + ": BrokenBarrierException");
				}
				
				cs.accept(tname + ": end...now=" + new Date());
			});
			
			td.start();
			
			tdarr[item] = td;
		});
		
		cs.accept("tdarr.length=" + tdarr.length);
		
		cs.accept(tnameMain + ": 1， parties=" + cb.getParties() + ", numWaiting=" + cb.getNumberWaiting());
		
		// 异常1：休眠10秒，在有等待线程时执行reset()
//		TimeUnit.SECONDS.sleep(10);
//		cb.reset();
		
		// 异常2：中断工作线程
		// 立即中断：子线程、主线程立即结束
		// 在工作线程await()时中断
		TimeUnit.SECONDS.sleep(5);
		IntStream.range(0, workNum).forEach(item->{
			// 中断每一个线程
			cs.accept(tnameMain + ": 中断线程-" + item);
			tdarr[item].interrupt();
		});

		cs.accept(tnameMain + ": isInterrupted=" + Thread.currentThread().isInterrupted());

		// 发生了异常：工作线程在await()时因为中断停止，cb被破坏，此时，main线程执行 await()发生异常 
		cs.accept(tnameMain + ": 等待 2， parties=" + cb.getParties() + ", numWaiting=" 
				+ cb.getNumberWaiting() + ", isBroken=" + cb.isBroken());
		// 重置可以修复cb，但要在新的业务中使用
//		if (cb.isBroken()) {
//			cb.reset();
//			cs.accept(tnameMain + ": after cb.reset() 2.1， parties=" + cb.getParties() 
//				+ ", numWaiting=" + cb.getNumberWaiting() + ", " + cb.isBroken());
//		}
		cb.await();
		
		cs.accept(tnameMain + ": 3， parties=" + cb.getParties() + ", numWaiting=" + cb.getNumberWaiting());
		
		cs.accept(tnameMain + ": end");
	}
	
	/**
	 * 基本使用
	 * @author ben
	 * @date 2021-09-15 09:18:19 CST
	 */
	public static void test1() {
		String tnameMain = Thread.currentThread().getName();
		cs.accept(tnameMain + ": start");
		
		final int workNum = 2;
		
		// 初始化参与方数量，后续无法更改
		CyclicBarrier cb = new CyclicBarrier(workNum + 1);
		
		// 启动工作线程
		IntStream.range(0, workNum).forEach(item->{
			new Thread(()->{
				String tname = Thread.currentThread().getName();
				cs.accept(tname + ": start");
				
				int slt = ThreadLocalRandom.current().nextInt(6);
				cs.accept(tname + ": sleep, slt=" + slt);
				try {
					TimeUnit.SECONDS.sleep(slt);
				} catch (InterruptedException e) {
					cs.accept(tname + ": InterruptedException 1");
				}
				
				cs.accept(tname + ": 任务结束, 等待其它...parties=" + cb.getParties() + ", numWaiting=" + cb.getNumberWaiting());
				
				try {
					// 子线程等待其它线程
					cb.await();
				} catch (InterruptedException e) {
					cs.accept(tname + ": InterruptedException 2");
				} catch (BrokenBarrierException e) {
					cs.accept(tname + ": BrokenBarrierException");
				}
				
				// 线程结束
				cs.accept(tname + ": end");
			}, "T" + item).start();
		});
		
		try {
			// 主线程等待子线程结束
			cs.accept(tnameMain + ": waiting...parties=" + cb.getParties() + ", numWaiting=" + cb.getNumberWaiting());
			// 使用wait()，就不需要使用join()等待了
			cb.await();
			
			// 休眠1秒 等待其它线程 完全结束
			TimeUnit.SECONDS.sleep(1);
			
			// 可以继续使用cb，开启新的任务...
			
		} catch (InterruptedException e) {
			cs.accept(tnameMain + ": InterruptedException");
		} catch (BrokenBarrierException e) {
			cs.accept(tnameMain + ": BrokenBarrierException");
		}
		
		cs.accept(tnameMain + ": end， parties=" + cb.getParties() + ", numWaiting=" + cb.getNumberWaiting());
	}
	
}
